package com.feihong.activity.service.impl;

import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.feihong.activity.configration.KafkaProducerConfiguration;
import com.feihong.activity.entity.EquipmentRequest;
import com.feihong.activity.entity.AreaInfo;
import com.feihong.activity.entity.JsonResult;
import com.feihong.activity.entity.Sensor;
import com.feihong.activity.exception.CustomException;
import com.feihong.activity.service.ThirdPartyService;
import com.feihong.common.service.DefaultServiceImpl;
import com.feihong.common.util.JsonObjectUtils;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author Filter
 */
@Service
public class ThirdPartyServiceImpl extends DefaultServiceImpl implements ThirdPartyService {
    private static final Long EXIT_TIME = 1L;
    private static final String TOPIC = "iot_device_log_chenfeihong";
    private static final String TOKEN_KEY = "Authenticator";
    @Resource
    private AreaInfo areaInfo;
    @Resource
    private RedisTemplate<String, String> redisTemplate;


    @Override
    public String equipmentDataReporting(List<Sensor> sensorList) throws JsonProcessingException {

        int size = sensorList.size();
        if (size == 0) {
            throw new CustomException("设备成功导入0条，无法进行上报");
        }
        StringBuilder result = new StringBuilder("设备共计 " + size + " 条,");
        //登录接口
        String token = login();
        //通过返回值判断是否成功
        int successCount = 0;
        int failureCount = 0;
        List<String> sensorIds = new ArrayList<>();
        //循环中请求
        for (Sensor sensor : sensorList) {
            ArrayList<Object> objects = new ArrayList<>();
            objects.add(sensor);
            HttpResponse response = HttpRequest.post(areaInfo.getBasicUrl()).header(TOKEN_KEY, token).body(JsonObjectUtils.getJsonByObject(objects)).execute();
            JsonResult jsonResult = (JsonResult) JsonObjectUtils.getObjectByString(response.body(), JsonResult.class);
            if ("200".equals(jsonResult.getCode())) {
                logger.info("设备成功上传："+response.body());
                successCount++;
            } else {
                logger.info("设备上传失败："+sensor.getSensorId());
                failureCount++;
                sensorIds.add(sensor.getSensorId());
            }
        }
        result.append("其中成功了 ").append(successCount).append(" 条,失败了 ").append(failureCount).append(" 条");
        logger.info("上传失败的设备id为： " + sensorIds);
        return result.toString();
    }

    @Override
    public <T> Integer pushKafkaByList(List<T> list) {
        Integer count = 0;
        for (T info : list) {
            try {
                String msg = JsonObjectUtils.getJsonByObject(info);
                KafkaProducerConfiguration kafkaProducer = new KafkaProducerConfiguration(TOPIC);
                kafkaProducer.getProducer().send(new ProducerRecord<String,String >(TOPIC,"chenMsg",msg));
                count++;
            } catch (JsonProcessingException e) {
                e.printStackTrace();
            }
        }
        return count;
    }

    private String login() {
        //先去redis里取令牌，若没有进行以下操作
        String authenticator = redisTemplate.opsForValue().get(TOKEN_KEY);
        if (authenticator == null) {
            HttpResponse execute = HttpRequest.get(areaInfo.getLoginUrl()).form("loginName", areaInfo.getLoginName()).form("password", areaInfo.getLoginPassword()).execute();
            authenticator = execute.header(TOKEN_KEY);
            if (authenticator != null) {
                redisTemplate.opsForValue().set(TOKEN_KEY, authenticator, EXIT_TIME, TimeUnit.HOURS);
            }
        }
        return authenticator;
    }
}
